package global;

import java.util.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.mapred.*;

public abstract class BaseDriver {

	private JobConf conf;
	private ArrayList<ArgumentParam> params;
	private String inputFolder;
	private String outputFolder;

	public BaseDriver(String[] args, String jobName) {
		params = fillParamsList();
		loadArgs(args);

		conf = createJobConf();
		GlobalJobConf.setCommonProps(conf, jobName);

		applyJobArgs();
		
		deleteOutputPath();

		doPreProcess();

		try {
			System.out.println("Executing job...");
			JobClient.runJob(conf);
			System.out.println("Job completed successfully!");
		}
		catch (Exception ex) {
			System.out.println("Error: Could not execute job.");
			System.out.println(ex.getMessage());
			System.exit(2);
		}

		doAfterProcess();
		System.out.println("SUCCESS!");
	}

	private void deleteOutputPath() {
		//delete the output hdfs directory (if it already exists) before starting the job
		System.out.println("Deleting output folder...");
		FileSystem hdfs = null;
		try {
			hdfs =  FileSystem.get(conf);
			try {
				Path path = new Path(outputFolder);
				hdfs.delete(path, true);
			}
			finally {
				hdfs.close();
			}
		}
		catch (Exception ex) {
			System.out.println("Warning: Could not delete output folder!");
			System.out.println(ex.getMessage());
		}
	}

	private void loadArgs(String[] args) {
		if (args.length > 0) {
			if (args[0].equalsIgnoreCase("/?")) {
				printUsage();
				System.exit(0);
			}
		}

		System.out.println("Loading command line arguments...");

		int requiredCount = 0;

		for (int i = 0; i < params.size(); i++) {
			if (params.get(i).isRequired()) {
				requiredCount++;
			}
		}

		if ((args.length < requiredCount) || (args.length > params.size())) {
			printUsage();
			System.exit(1);
		}

		for (int i = 0; i < args.length; i++) {
			switch (params.get(i).getType()) {
			case GENERAL_INT:
				try {
					Integer.parseInt(args[i]);
				}
				catch (Exception ex) {
					printParseError(params.get(i).getName(), "Integer");
				}
				break;
			default:
				break;
			}
			params.get(i).setCurrentValue(args[i]);
		}

		for (int i = 0; i < params.size(); i++) {
			System.out.println("Using value: '" + params.get(i).getCurrentValue() + "' for argument <" + params.get(i).getName() + ">");
		}
	}

	private void applyJobArgs() {
		inputFolder = getArgByType(ArgumentType.INPUT_FOLDER).getCurrentValue();
		outputFolder = getArgByType(ArgumentType.OUTPUT_FOLDER).getCurrentValue();

		FileInputFormat.setInputPaths(conf, new Path(inputFolder));
		FileOutputFormat.setOutputPath(conf, new Path(outputFolder));

		System.out.println("Sending arguments to M/R job...");
		for (int i = 0; i < params.size(); i++) {
			if (params.get(i).getShouldBePassedToJob()) {
				switch (params.get(i).getType()) {
				case GENERAL_INT:
					conf.setInt(params.get(i).getName(), Integer.parseInt(params.get(i).getCurrentValue()));
					break;
				case GENERAL_STRING:
					conf.set(params.get(i).getName(), params.get(i).getCurrentValue());
					break;
				default:
					break;
				}
			}
		}
	}

	protected ArgumentParam getArgByType(ArgumentType type) {
		ArgumentParam res = null;

		for (int i = 0; i < params.size(); i++) {
			if (params.get(i).getType() == type) {
				res = params.get(i);
			}
		}

		return res;
	}

	private void printParseError(String paramName, String paramType) {
		System.out.println("Could not successfully parse argument <" + paramName + ">. Expecting type: " + paramType + ".");
		printUsage();
	}

	private void printUsage() {
		System.out.print("Usage: ");
		for (int i = 0; i < params.size(); i++) {
			if (!params.get(i).isRequired()) {
				System.out.print("[");
			}

			System.out.print("<" + params.get(i).getName() + ">");

			if (!params.get(i).isRequired()) {
				System.out.print(" = " + params.get(i).getDefaultValue() + "]");
			}

			System.out.print(" ");
		}
		System.out.println();
		printExtraUsage();
	}

	protected ArrayList<ArgumentParam> getParamsList() {
		return params;
	}

	protected JobConf getJobConf() {
		return conf;
	}

	protected String getInputFolder() {
		return inputFolder;
	}

	protected String getOutputFolder() {
		return outputFolder;
	}


	//IMPORTANT: Required arguments should appear in the list prior to all optional arguments
	protected abstract ArrayList<ArgumentParam> fillParamsList();

	//To be used for setting the appropriate job classes
	protected abstract JobConf createJobConf();

	//To be used by derived classes for pre-job actions
	protected void doPreProcess() { }

	//To be used by derived classes for after-job actions
	protected void doAfterProcess() { }


	//Any derived class that needs to print extra usage info, should override this function
	protected void printExtraUsage() { }
}
